package org.cancer.service

import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode}
import org.cancer.util.SparkUtil
import org.apache.spark.sql.functions._


class LungCancerIncidenceService$Laurel {

  def dataAnalysis(data: DataFrame): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._ //隐式转换

    val filterDF: DataFrame = data.filter($"肿瘤类型" === "肺癌")
      .withColumn("省份人口数（万人）", $"省份人口数（万人）".cast("double")) // 确保人口数列是double类型

    val cancerCountDF: DataFrame = filterDF.groupBy($"省份")
      .agg(count("*").alias("肺癌患者数"))

    // 5. 获取各省份人口数（每个省份取第一条记录的人口数作为该省总人口数）
    val populationDF: Dataset[Row] = filterDF
      .select($"省份", $"省份人口数（万人）")
      .dropDuplicates("省份")

    // 计算各省份肺癌发病率（每万人中的患病人数）
    val incidenceDF = cancerCountDF
      .join(populationDF, Seq("省份"), "inner")
      .withColumn("肺癌发病率（每万人）", ($"肺癌患者数" / $"省份人口数（万人）") * 10000)
      .withColumn("肺癌发病率（每万人）", format_number($"肺癌发病率（每万人）", 2)) // 保留两位小数
      .select($"省份", $"肺癌患者数", $"省份人口数（万人）", $"肺癌发病率（每万人）")
      .orderBy($"肺癌发病率（每万人）".desc)

    //存入数据库
    incidenceDF.write
      .format("jdbc") //指定数据源 链接mysql odbc  ORACLE
      .option("url", "jdbc:mysql://node1:3306/cancer_patients") //链接本地 jdbc:mysql://localhost:3306/ 本地的数据库
      .option("driver", "com.mysql.cj.jdbc.Driver") //mysql 8.0以上：com.mysql.cj.jdbc.Driver
      .option("user", "root") //用户名
      .option("password", "123456") //密码
      .option("dbtable", "lungCancer_laurel") //表名
      .mode(SaveMode.Append) //当数据不存在的时候会自动创建
      .save()
  }
}
